package belf.migrate.api.sink.jdbc;

import belf.migrate.api.sink.SinkEvent;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Base64;
import java.util.Iterator;
import java.util.List;

/**
 * 从redis队列中获取到的精简后的CDC JSON消息，转换为可插入关系数据库的事件
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class JdbcSinkEvent implements SinkEvent {
    private static final long serialVersionUID = 1L;
    /** 主键字段，可以是多个。对于无主键表，总是null */
    private List<KeyValue> keys;
    /** 事件类型，目前仅仅处理INSERT、UPDATE、DELETE */
    private EventType eventType;
    /** 对应的数据库名或schema名 */
    private String database;
    /** 对应的表名 */
    private String table;

//    目前还没处理DDL事件
//    private String ddl;

    /** 对应的行修改前的值。对于INSERT事件，总是null */
    private List<KeyValue> before;
    /** 对应的行修改后的值。对于DELETE事件，总是null */
    private List<KeyValue> after;

    public static JdbcSinkEvent buildFromJson(String jsonString) throws JsonProcessingException {
        ObjectMapper mapper = new ObjectMapper();
        ObjectNode rootNode = (ObjectNode)mapper.readTree(jsonString);

        EventType eventType = op(rootNode.get("op").textValue());
        List<KeyValue> keys = toFields(rootNode.get("key"));
//        if (keys != null && keys.size() > 0) {
//            eventType = EventType.CREATE;
//        }
        return new JdbcSinkEventBuilder().keys(keys)
                .after(toFields(rootNode.get("after")))
                .before(toFields(rootNode.get("before")))
                .database(rootNode.get("db").textValue())
                .table(rootNode.get("table").textValue())
                .eventType(eventType)
                .build();

    }

    private static EventType op(String value) {
        switch (value) {
            case "c": return EventType.CREATE;
            case "u": return EventType.UPDATE;
            case "d": return EventType.DELETE;
            case "r": return EventType.READ;
            case "ddl": return EventType.DDL;
            default:return EventType.CREATE;
        }

    }

    public static List<KeyValue> toFields(JsonNode node) {
        if (null == node) {
            return null;
        }

        List<KeyValue> fields = new ArrayList<>();
        Iterator<String> it = node.fieldNames();
        while (it.hasNext()) {
            String key = it.next();
            JsonNode valueNode = node.get(key);
            if (valueNode.isNull()) {
                fields.add(new KeyValue(key, null));
            } else if (valueNode.isBigDecimal()) {
                BigDecimal bigDecimal = new BigDecimal(valueNode.asText());
                fields.add(new KeyValue(key, bigDecimal));
            } else if (valueNode.isBigInteger()) {
                BigInteger bigInteger = BigInteger.valueOf(valueNode.asLong());
                fields.add(new KeyValue(key, bigInteger));
            } else if (valueNode.isBoolean()) {
                Boolean boolValue = valueNode.asBoolean();
                fields.add(new KeyValue(key, boolValue));
            } else if(valueNode.isBinary()) { // BASE64编码的二进制内容
                byte[] decodedBytes = Base64.getDecoder().decode(valueNode.textValue());
                fields.add(new KeyValue(key, decodedBytes));
            } else if (valueNode.isDouble()) {
                Double aDouble = valueNode.asDouble();
                fields.add(new KeyValue(key, aDouble));
            } else if (valueNode.isFloat()) {
                Float aFloat = Float.parseFloat(valueNode.asText());
                fields.add(new KeyValue(key, aFloat));
            } else if (valueNode.isInt()) {
                Integer aInt = valueNode.asInt();
                fields.add(new KeyValue(key, aInt));
            } else if (valueNode.isLong() ) {
                Long aLong = valueNode.asLong();
                fields.add(new KeyValue(key, aLong));
            } else if (valueNode.isShort() ) {
                Short aShort = Short.parseShort(valueNode.textValue());
                fields.add(new KeyValue(key, aShort));
            } else if (valueNode.isTextual() ) {
                fields.add(new KeyValue(key, valueNode.textValue()));
            } else {
                fields.add(new KeyValue(key, valueNode));
            }
        }
        return fields;
    }

    public enum EventType {

        CREATE(1, "INSERT"),
        UPDATE(2, "UPDATE"),
        DELETE(3, "DELETE"),
        //对应标记为"r"的事件，如果key是null，before是null，after有值，则是无主键表的插入
        READ(4, "READ"),
        DDL(5, "DDL");

        private final int code;
        private final String message;

        EventType(int code, String message) {
            this.code = code;
            this.message = message;
        }

        public int getCode() {
            return code;
        }

        public String getMessage() {
            return message;
        }
    }

}
